package com.intct.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.common.Constant;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author gufg
 * @since 2025-09-18 15:02
 */
public class DimBoradcatProcessFuncImpl extends BroadcastProcessFunction<String, String, String> {
    // 类变量
    private MapStateDescriptor mapStateDescriptor;

    private Map<String, String> configMap = new HashMap<>();

    // 类构造方法
    public DimBoradcatProcessFuncImpl(MapStateDescriptor mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        // JDBC -- read all travel_config.dim_table_config
        /*
         url: jdbc:mysql://cdh-node:13306/travel_config
         username : root
         password : Test_090110
         驱动类型名称 : com.mysql.cj.jdbc.Driver
         */
        Class.forName("com.mysql.cj.jdbc.Driver");
        Connection jdbcConn = DriverManager.getConnection("jdbc:mysql://cdh-node:13306/travel_config", "root", "Test_090110");

        PreparedStatement preparedStatement = jdbcConn.prepareStatement("select * from travel_config.dim_table_config");
        ResultSet resultSet = preparedStatement.executeQuery();

        while (resultSet.next()) {
            String key = resultSet.getString("source_table");
            JSONObject jsonObj = new JSONObject();

            JSONObject dataObj = new JSONObject();
            dataObj.put("source_table", resultSet.getString("source_table"));
            dataObj.put("sink_table", resultSet.getString("sink_table"));
            dataObj.put("sink_family", resultSet.getString("sink_family"));
            dataObj.put("sink_columns", resultSet.getString("sink_columns"));
            dataObj.put("sink_row_key", resultSet.getString("sink_row_key"));

            jsonObj.put("op", "r");
            jsonObj.put(Constant.FLINK_CDC_MYSQL_JSON_AFTER_KEY, dataObj);

            configMap.put(key, jsonObj.toJSONString());
        }

        resultSet.close();
        preparedStatement.close();
        jdbcConn.close();
    }

    // 主流
    @Override
    public void processElement(String jsonStr,
                               BroadcastProcessFunction<String, String, String>.ReadOnlyContext ctx,
                               Collector<String> out) throws Exception {
        // 主流  == {"op":"r","file":"","pos":"0","after":{"code":341723,"name":"安徽省池州市青阳县"},"db":"travel","table":"area_info"}
        JSONObject jsonObj = JSON.parseObject(jsonStr);

        // 表名称 -- 业务库表名
        String tableName = jsonObj.getString(Constant.FLINK_CDC_MYSQL_JSON_TABLE_KEY);

        // 获取MapState状态
        ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        // 主流与广播流关联
        String stateValue = broadcastState.get(tableName) == null ? configMap.get(tableName) : broadcastState.get(tableName);
        JSONObject broadcastJsonObj = JSON.parseObject(stateValue);

        // 问题：是在Flink多流关联时，流数据先到与后到，哪个流数据先到不确定，所以产生无数据或者转换失败
        // System.out.println("主流："+ jsonObj.toJSONString() + "  广播流："+ JSON.toJSONString(broadcastState));

        // 判断是维度表
        if (stateValue != null) {
            // 保存HBase  ？？？
            // 表名、字段、rowkey、列族
            // put '表名', 'rowkey',         '列族：列限定符', '值'
            //      广     主(值)、广(字段名)   广：广,          主
            // 主流与广播流字段不一致

            JSONObject midObj = new JSONObject();

            // 获取值
            // rowkey
            JSONObject broadcastAfterObj = null;
            if (broadcastJsonObj.getJSONObject(Constant.FLINK_CDC_MYSQL_JSON_AFTER_KEY) == null) {
                JSONObject configMapObj = JSON.parseObject(JSON.toJSONString(configMap));
                // TODO 需要判断配置表(广播流)中对数据删除时，只有before里有数据
                broadcastAfterObj = configMapObj.getJSONObject(Constant.FLINK_CDC_MYSQL_JSON_AFTER_KEY);
            } else {
                broadcastAfterObj = broadcastJsonObj.getJSONObject(Constant.FLINK_CDC_MYSQL_JSON_AFTER_KEY);
            }

            // 表名
            String sinkTable = broadcastAfterObj.getString("sink_table");
            midObj.put("table", sinkTable);

            String sinkRowKeyColName = broadcastAfterObj.getString("sink_row_key");
            JSONObject afterObj = jsonObj.getJSONObject(Constant.FLINK_CDC_MYSQL_JSON_AFTER_KEY);
            String rowkeyValue = afterObj.getString(sinkRowKeyColName);

            midObj.put("rowkey", rowkeyValue);

            // 列族
            String sinkFamily = broadcastAfterObj.getString("sink_family");
            midObj.put("sink_family", sinkFamily);

            // 主流与广播流字段不一致
            String sinkColumns = broadcastAfterObj.getString("sink_columns");
            List<String> sinkColumnList = Arrays.asList(sinkColumns.split(","));

            JSONObject tmpObj = new JSONObject();
            for (String column : sinkColumnList) {
                if (sinkColumnList.contains(column)) {
                    tmpObj.put(column, afterObj.getString(column));
                }
            }

            // 组装成一个Json string  {"table":"sink_order_state", "rowkey" :"10001", "sink_family":"info", "after":{"state":"10001", "name":"已上车"}}
            midObj.put("after", tmpObj);

            out.collect(midObj.toJSONString());
        }

    }

    // 广播流处理(mysql数仓配置数据库)
    @Override
    public void processBroadcastElement(String jsonStr,
                                        BroadcastProcessFunction<String, String, String>.Context ctx,
                                        Collector<String> out) throws Exception {
        // 获取MapState状态
        BroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        // 广播流数据 json string -> json object
        JSONObject jsonObj = JSON.parseObject(jsonStr);

        // 获取source key中的db table -->
        JSONObject sourceObj = jsonObj.getJSONObject(Constant.FLINK_CDC_MYSQL_JSON_SOURCE_KEY);

        // 删除JSON中多出的KEY
        jsonObj.remove(Constant.FLINK_CDC_MYSQL_JSON_SOURCE_KEY);
        jsonObj.remove(Constant.FLINK_CDC_MYSQL_JSON_TS_MS_KEY);
        jsonObj.remove(Constant.FLINK_CDC_MYSQL_JSON_TTRASACTION_KEY);

        // 操作类型
        String op = jsonObj.getString(Constant.FLINK_CDC_MYSQL_JSON_OP_KEY);

        // 如果MySQL配置中删除数据
        if ("d".equals(op)) {
            JSONObject beforeObj = jsonObj.getJSONObject(Constant.FLINK_CDC_MYSQL_JSON_BEFORE_KEY);
            broadcastState.remove(beforeObj.getString("source_table"));
            configMap.remove(beforeObj.getString("source_table"));
        } else {
            JSONObject afterObj = jsonObj.getJSONObject(Constant.FLINK_CDC_MYSQL_JSON_AFTER_KEY);
            broadcastState.put(afterObj.getString("source_table"), jsonObj.toJSONString());
            configMap.put(afterObj.getString("source_table"), jsonObj.toJSONString());
        }
    }

}
